package com.ndp.fb.rdb.mongo.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.*;
import com.ndp.fb.constant.MongoConstants;
import com.ndp.fb.enums.adresource.ThirdPicReviewStatus;
import com.ndp.fb.model.AdResourceComment;
import com.ndp.fb.model.AdResourceStats;
import com.ndp.fb.model.business.vo.stats.AdResourceStatsQueryParam;
import com.ndp.fb.mongo.api.AdResourceStatsMongoService;
import com.ndp.fb.mongo.api.IMongodbService;
import com.ndp.fb.mongo.model.UserAdResourceStats;
import com.ndp.fb.util.DateUtil;
import com.ndp.fb.util.ListUtil;
import com.ndp.fb.util.StringUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.text.SimpleDateFormat;
import java.util.*;

@Service
public class AdResourceStatsMongoServiceImpl implements AdResourceStatsMongoService {
    // 可以查询的业务字段
    private static String[] BUSINESS_FIELDS = {"mongodbId", "createTime", "height", "isDelete", "litImgURL", "litNfsId", "name", "nfsId", "orgId", "resourceType", "tags", "uploadStatus", "userIdSet", "visitPath", "width", "reviewStatus", "reviewer", "reviewTime", "adReviewFeedback", "machineCnt", "comments", "md5", "imageHash"};
    private static String[] MAYBE_NULL_FIELDS = {"totoalPageViews", "numOfPageViews", "peopleNumOfPageViews"};
    // 可以进行SUM的字段
    private static String[] SUM_FIELDS = {"clicks", "impressions", "results", "socialClicks", "socialImpressions", "socialSpent", "socialUniqueClicks", "socialUniqueImpressions", "spent", "uniqueClicks", "uniqueImpressions"};
    @Autowired
    private IMongodbService mongodbService;
    private Logger logger = LoggerFactory.getLogger(getClass());
    @Autowired
    private DB adResourceMongoDBClient;

    public void batchSaveByWhereWithoutHistory(List<AdResourceStats> adResourceStatsList, String[] where) {
        if (checkEmpty(adResourceStatsList)) return;
        mongodbService.batchSaveOrUpdateWithoutHistory(adResourceMongoDBClient, MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY, JSONArray.toJSONString(adResourceStatsList), where, false);
        logger.info("-----向mongoDB：" + MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY + "集合插入数据完成");
    }

    private boolean checkEmpty(List<AdResourceStats> adResourceStatsList) {
        if (CollectionUtils.isEmpty(adResourceStatsList)) {
            logger.info("AdResource统计数据入库列表为空，退出入库操作");
            return true;
        }
        return false;
    }

    @Override
    public void batchIncreaseByWhereWithoutHistory(List<AdResourceStats> adResourceStatsList, String[] where) {
        if (checkEmpty(adResourceStatsList)) return;
        mongodbService.batchSaveOrIncreaseStatsData(adResourceMongoDBClient, MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY, JSONArray.toJSONString(adResourceStatsList), where);
        logger.info("-----向mongoDB：" + MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY + "集合插入数据完成");
    }

    /**
     * 根据MongodbId查询
     *
     * @param mongodbId mongodbId
     * @return 查询结果
     */
    @Override
    public AdResourceStats findByMongodbId(String mongodbId) {
        if (StringUtils.isEmpty(mongodbId)) {
            throw new IllegalArgumentException("参数错误，mongodbId不能为空。");
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("mongodbId", mongodbId);
        DBObject dbObject = collection.findOne(query);
        if (dbObject == null) {
            return null;
        } else {
            return JSONObject.parseObject(JSON.toJSONString(dbObject), AdResourceStats.class);
        }
    }

    @Override
    public List<AdResourceStats> findALL1970() {
        List<AdResourceStats> list = new ArrayList<>();
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("day", "1970-01-01");

        DBCursor dbObjects = collection.find(query);
        while (dbObjects.hasNext()) {
            DBObject next = dbObjects.next();
            AdResourceStats adResourceStats = JSONObject.parseObject(JSON.toJSONString(next), AdResourceStats.class);
            list.add(adResourceStats);
        }

        return list;
    }

    /**
     * 根据MongodbId查询
     *
     * @param mongodbIds
     * @return 查询结果
     */
    @Override
    public List<AdResourceStats> findByMongodbIds(List<String> mongodbIds) {
        if (ListUtil.isEmpty(mongodbIds)) {
            throw new IllegalArgumentException("参数错误，mongodbIds不能为空。");
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        BasicDBList basicDBList = new BasicDBList();
        basicDBList.addAll(mongodbIds);
        query.put("mongodbId", new BasicDBObject("$in", basicDBList));
        DBCursor cursor = collection.find(query).addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<AdResourceStats> adResourceStatses = new ArrayList<>();
        try {
            while (cursor.hasNext()) {
                DBObject next = cursor.next();
                AdResourceStats adResourceStats = JSONObject.parseObject(JSON.toJSONString(next), AdResourceStats.class);
                adResourceStatses.add(adResourceStats);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return adResourceStatses;
    }

    /**
     * @param path
     * @return
     * @discription
     * @author giser.yang
     * 创建时间 2015年7月28日 上午11:43:23
     * @see com.ndp.fb.mongo.api.AdResourceStatsMongoService#findMd5ByVistPath(java.lang.String)
     */

    @Override
    public String findMd5ByVistPath(String path) {
        if (StringUtils.isEmpty(path)) {
            logger.info("vistpath为空,不查询");
            return null;
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("visitPath", path);
        query.put("md5", new BasicDBObject("$ne", null));
        DBObject dbObject = collection.findOne(query);
        if (dbObject == null) {
            return null;
        }
        if (dbObject.get("md5") == null) {
            return null;
        }
        return dbObject.get("md5").toString();
    }

    /**
     * 根据图片的md5汇总素材数据-明细数据
     *
     * @param adResourceStatsQueryParam 查询参数
     *                                  userIdSet           根据图片所有者检索的过滤条件（非必须项）
     *                                  accountIdSet        图片所有者或者公司的所有广告账号（非必须项，如果需要统计数据则此项为必须）
     *                                  resourceType        素材类型（非必须项），取值范围CommonConstants.RESOURCE_TYPE_IMAGE和CommonConstants.RESOURCE_TYPE_VIDEO
     *                                  orgId               根据公司检索的过滤条件（非必须项）
     *                                  tags                标签过滤条件（非必须项）
     *                                  startDay            统计开始日（非必须项，如果需要统计数据则此项为必须），包含该天
     *                                  endDay              统计截止日（非必须项，如果需要统计数据则此项为必须），包含该天
     *                                  sortList            排序字段及排序类型列表（支持多字段排序），List中每个元素是一个Map，Map的Key是排序字段，Map的Value是排序类型（1为升序，-1为降序）
     *                                  page                查询第几页（非必须项），从1开始，如果分页查询时为必须项
     *                                  pageSize            每页的数量（非必须项），如果分页查询时为必须项
     *                                  onlyNew             是否只统计新建广告的流水
     *                                  returnStatsFields   需要返回的字段（必须项），取值范围["cost","ctr","cr","clicks","impressions","results","socialClicks","socialImpressions","socialSpent","socialUniqueClicks","socialUniqueImpressions","spent","uniqueClicks","uniqueImpressions","mongodbId","createTime","height","isDelete","litImgURL","litNfsId","name","nfsId","orgId","resourceType","tags","uploadStatus","userIdSet","visitPath","width"]
     * @return 统计结果
     */
    @Override
    public Map<String, Object> groupByImageHash(AdResourceStatsQueryParam adResourceStatsQueryParam) {
        Set<String> returnFields = adResourceStatsQueryParam.getReturnFields();
        if (CollectionUtils.isEmpty(returnFields)) {
            throw new IllegalArgumentException("参数错误，returnFields为必须项。");
        }

        List<DBObject> pipeline = new ArrayList<DBObject>();

        // 生成day过滤对象
        List<DBObject> combineDayDBObject = genDayFilterObject(adResourceStatsQueryParam);

        // 第一次Match，主要是根据day进行过滤
        BasicDBObject matchDBObject1 = new BasicDBObject();
        matchDBObject1.put("$match", new BasicDBObject("$or", combineDayDBObject));
        pipeline.add(matchDBObject1);

        // 第一次排序，将1970的数据都放在前面，这样就可以取到业务数据
        BasicDBObject sortDBObject1 = new BasicDBObject();
        sortDBObject1.put("$sort", new BasicDBObject("day", 1));
        pipeline.add(sortDBObject1);

        // 第一次分组，根据imageHash或者fbConnectionObjectId进行分组
        BasicDBObject groupChildDBObject1 = new BasicDBObject();
        // 第一次分组默认按照imageHash分组，如果需要的话还可以按照md5和应用分组
        BasicDBObject idObject = new BasicDBObject("imageHash", "$imageHash");
        groupChildDBObject1.put("_id", idObject);
        addBusinessFieldToGroupObject(groupChildDBObject1, "day");

        Boolean isGroupByFbConnectionObjectId = adResourceStatsQueryParam.getIsGroupByFbConnectionObjectId();
        // 第一次Project，计算ctr,cr,cost等值
        BasicDBObject projectChildDBObject1 = new BasicDBObject()
                .append("imageHash", "$_id.imageHash");
        if (null != isGroupByFbConnectionObjectId && isGroupByFbConnectionObjectId) {
            idObject.put("fbConnectionObjectId", "$fbConnectionObjectId");
            projectChildDBObject1.put("fbConnectionObjectId", "$_id.fbConnectionObjectId");
        }

        // 第二次分组，将所有数据汇总到一起，计算总条数
        BasicDBObject groupChildDBObject2 = new BasicDBObject()
                .append("_id", null)
                .append("totalCount", new BasicDBObject("$sum", 1))
                .append("adResourceStatsList", new BasicDBObject("$push", "$$CURRENT"));

        if (!CollectionUtils.isEmpty(returnFields)) {
            for (String returnField : returnFields) {
                if (Arrays.asList(BUSINESS_FIELDS).contains(returnField)) {
                    addBusinessFieldToGroupObject(groupChildDBObject1, returnField);
                    addFieldToTargetObject(projectChildDBObject1, returnField, "$" + returnField);
                } else if (Arrays.asList(MAYBE_NULL_FIELDS).contains(returnField)) {
                    addBusinessFieldToGroupObject(groupChildDBObject1, returnField);
                    addFieldToTargetObject(projectChildDBObject1, returnField, new BasicDBObject("$ifNull", new Object[]{"$" + returnField, 0}));
                } else if (Arrays.asList(SUM_FIELDS).contains(returnField)) {
                    addSumFieldToGroupObject(groupChildDBObject1, returnField);
                    addFieldToTargetObject(projectChildDBObject1, returnField, "$" + returnField);
                } else if ("cost".equals(returnField)) {
                    addSumFieldToGroupObject(groupChildDBObject1, "spent");
                    addSumFieldToGroupObject(groupChildDBObject1, "results");
                    addFieldToTargetObject(projectChildDBObject1, "spent", "$spent");
                    addFieldToTargetObject(projectChildDBObject1, "results", "$results");
                    BasicDBObject costDBObject = genConditionDBObject(new BasicDBObject("$eq", new Object[]{"$results", 0}), "$spent", new BasicDBObject("$divide", new Object[]{"$spent", "$results"}));
                    addFieldToTargetObject(projectChildDBObject1, "cost", new BasicDBObject("$cond", costDBObject));
                } else if ("ctr".equals(returnField)) {
                    addSumFieldToGroupObject(groupChildDBObject1, "clicks");
                    addSumFieldToGroupObject(groupChildDBObject1, "impressions");
                    addFieldToTargetObject(projectChildDBObject1, "clicks", "$clicks");
                    addFieldToTargetObject(projectChildDBObject1, "impressions", "$impressions");
                    // CTR
                    BasicDBObject ctrDBObject = genConditionDBObject(new BasicDBObject("$eq", new Object[]{"$impressions", 0}), 0, new BasicDBObject("$divide", new Object[]{"$clicks", "$impressions"}));
                    addFieldToTargetObject(projectChildDBObject1, "ctr", new BasicDBObject("$cond", ctrDBObject));
                } else if ("cr".equals(returnField)) {
                    addSumFieldToGroupObject(groupChildDBObject1, "results");
                    addSumFieldToGroupObject(groupChildDBObject1, "clicks");
                    addFieldToTargetObject(projectChildDBObject1, "results", "$results");
                    addFieldToTargetObject(projectChildDBObject1, "clicks", "$clicks");
                    // CR
                    BasicDBObject crDBObject = genConditionDBObject(new BasicDBObject("$eq", new Object[]{"$clicks", 0}), 0, new BasicDBObject("$divide", new Object[]{"$results", "$clicks"}));
                    addFieldToTargetObject(projectChildDBObject1, "cr", new BasicDBObject("$cond", crDBObject));
                }
            }
        }
        pipeline.add(new BasicDBObject("$group", groupChildDBObject1));

        // 第二次Match，主要是进行应用过滤以及统计值大小比较过滤
        pipeline.add(new BasicDBObject("$match", new BasicDBObject("day", "1970-01-01")));

        // 第一次Project，计算ctr,cr,cost等值
        pipeline.add(new BasicDBObject("$project", projectChildDBObject1));
        List<Map<String, Integer>> sortList = adResourceStatsQueryParam.getSortList();
        if (!CollectionUtils.isEmpty(sortList)) {
            BasicDBObject sortDBObject = new BasicDBObject();
            for (Map<String, Integer> sort : sortList) {
                Set<Map.Entry<String, Integer>> entries = sort.entrySet();
                for (Map.Entry<String, Integer> entry : entries) {
                    sortDBObject.put(entry.getKey(), entry.getValue());
                }
            }
            pipeline.add(new BasicDBObject("$sort", sortDBObject));
        }

        // 第二次分组，将所有数据汇总到一起，计算总条数
        pipeline.add(new BasicDBObject("$group", groupChildDBObject2));

        // 用来进行分页
        BasicDBObject projectChildDBObject2 = new BasicDBObject()
                .append("summary", new BasicDBObject("totalCount", "$totalCount"));

        Integer page = adResourceStatsQueryParam.getPage();
        Integer pageSize = adResourceStatsQueryParam.getPageSize();
        if (null != page && 0 != page && null != pageSize && 0 != pageSize) {
            projectChildDBObject2.put("adResourceStatsList", new BasicDBObject("$slice", new Object[]{"$adResourceStatsList", (page - 1) * pageSize, pageSize}));
        } else {
            projectChildDBObject2.put("adResourceStatsList", 1);
        }
        pipeline.add(new BasicDBObject("$project", projectChildDBObject2));

        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(pipeline));

        Long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY).aggregate(pipeline, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        Map<String, Object> result = null;
        try {
            if (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                result = JSONObject.parseObject(JSON.toJSONString(dbObject), Map.class);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        logger.info("获取明细行耗时：" + (System.currentTimeMillis() - start));
        return result;
    }

    /**
     * 汇总统计数据时生成day相关的过滤条件对象
     *
     * @param adResourceStatsQueryParam 查询参数
     * @return 生成好的过滤对象
     */
    private List<DBObject> genDayFilterObject(AdResourceStatsQueryParam adResourceStatsQueryParam) {
        Boolean onlyNew = adResourceStatsQueryParam.getOnlyNew();
        BasicDBObject statsDayDBObject = new BasicDBObject();
        String startDay = adResourceStatsQueryParam.getStartDay();
        String endDay = adResourceStatsQueryParam.getEndDay();
        if (null != onlyNew && onlyNew) {
            // 预先算好的数据
            statsDayDBObject.put("day", "1970-01-02");
            long dayDiff = 1;
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            try {
                dayDiff = DateUtil.dayDiff(simpleDateFormat.parse(startDay), simpleDateFormat.parse(endDay));
            } catch (Exception e) {
                logger.error("时间转换失败，startDay={},endDay={}", startDay, endDay);
                // 使用最近一天的数据
            }
            statsDayDBObject.put("dayRange", dayDiff);
        } else {
            BasicDBObject sinceAndUntilDayDBObject = new BasicDBObject();
            if (StringUtils.isNotEmpty(startDay)) {
                sinceAndUntilDayDBObject.put("$gte", startDay);
            }
            if (StringUtils.isNotEmpty(endDay)) {
                sinceAndUntilDayDBObject.put("$lte", endDay);
            }
            if (!sinceAndUntilDayDBObject.isEmpty()) {
                statsDayDBObject.put("day", sinceAndUntilDayDBObject);
            }
        }
        Set<Long> fbConnectionObjectIds = adResourceStatsQueryParam.getFbConnectionObjectIds();
        if (!CollectionUtils.isEmpty(fbConnectionObjectIds)) {
            statsDayDBObject.put("fbConnectionObjectId", new BasicDBObject("$in", fbConnectionObjectIds));
        }

        Set<Long> accountIdSet = adResourceStatsQueryParam.getAccountIdSet();
        if (!CollectionUtils.isEmpty(accountIdSet)) {
            statsDayDBObject.put("accountId", new BasicDBObject("$in", accountIdSet));
        }

        String breakdowns = adResourceStatsQueryParam.getBreakdowns();
        if (StringUtils.isNotEmpty(breakdowns)) {
            statsDayDBObject.put("breakdowns", breakdowns);
        }

        // 1970-01-01
        BasicDBObject businessDayDBObject = new BasicDBObject();

        businessDayDBObject.put("day", "1970-01-01");
        Long orgId = adResourceStatsQueryParam.getOrgId();
        if (null != orgId && 0 != orgId) {
            businessDayDBObject.put("orgId", orgId);
        }
        Set<String> userIdSet = adResourceStatsQueryParam.getUserIdSet();
        if (!CollectionUtils.isEmpty(userIdSet)) {
            businessDayDBObject.put("userIdSet", new BasicDBObject("$in", userIdSet));
        }
        Integer reviewStatus = adResourceStatsQueryParam.getReviewStatus();
        if (reviewStatus != null) {
            businessDayDBObject.put("reviewStatus", reviewStatus);
        }

        Integer isDelete = adResourceStatsQueryParam.getIsDelete();
        if (isDelete != null) {
            businessDayDBObject.put("isDelete", isDelete);
        }

        Integer from = adResourceStatsQueryParam.getFrom();
        if (from != null) {
            businessDayDBObject.put("from", from);
        }
        Map<String, Set<Object>> tags = adResourceStatsQueryParam.getTags();
        if (!CollectionUtils.isEmpty(tags)) {
            for (Map.Entry<String, Set<Object>> tag : tags.entrySet()) {
                String key = tag.getKey();
                Set<Object> value = tag.getValue();
                businessDayDBObject.put("tags." + key, new BasicDBObject("$in", value));
            }
        }

        // ResourceType过滤
        String resourceType = adResourceStatsQueryParam.getResourceType();
        if (StringUtils.isNotEmpty(resourceType)) {
            businessDayDBObject.put("resourceType", resourceType);
        }

        // 组合day过滤对象
        List<DBObject> combineDayDBObject = new ArrayList<>();
        if (!statsDayDBObject.isEmpty()) {
            combineDayDBObject.add(statsDayDBObject);
        }
        combineDayDBObject.add(businessDayDBObject);
        return combineDayDBObject;
    }

    /**
     * 往给定的GroupDBObject中追加业务的字段（取第一条数据的字段）
     *
     * @param groupDBObject 需要追加的GroupDBObject
     * @param fieldName     追加字段名称
     */
    private void addBusinessFieldToGroupObject(DBObject groupDBObject, String fieldName) {
        groupDBObject.put(fieldName, new BasicDBObject("$first", "$" + fieldName));
    }

    /**
     * 往给定的GroupDBObject中追加sum的字段
     *
     * @param groupDBObject 需要追加的GroupDBObject
     * @param fieldName     追加字段名称
     */
    private void addSumFieldToGroupObject(DBObject groupDBObject, String fieldName) {
        groupDBObject.put(fieldName, new BasicDBObject("$sum", "$" + fieldName));
    }

    /**
     * 往给定的targetObject中追加字段
     *
     * @param targetObject 需要追加的ProjectDBObject
     * @param fieldName    追加字段名称
     * @param value        追加字段对应的值
     */
    private void addFieldToTargetObject(DBObject targetObject, String fieldName, Object value) {
        targetObject.put(fieldName, value);
    }

    /**
     * 生成一个条件表达式DBObject
     *
     * @param ifObject   ifObject
     * @param thenObject thenObject
     * @param elseObject elseObject
     * @return
     */
    private BasicDBObject genConditionDBObject(Object ifObject, Object thenObject, Object elseObject) {
        BasicDBObject costDBObject = new BasicDBObject();
        costDBObject.put("if", ifObject);
        costDBObject.put("then", thenObject);
        costDBObject.put("else", elseObject);
        return costDBObject;
    }

    @Override
    public void updateNameAndTags(String mongodbId, String fileName, Integer reviewStatus, String reviewer, Map<String, List> tags, boolean overrideTags) {
        DBObject query = new BasicDBObject("mongodbId", mongodbId);
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        if (overrideTags) {
            DBObject mongoTags = new BasicDBObject();

            if (tags != null && tags.size() > 0) {
                mongoTags.put("tags", tags);
            }

            if (StringUtils.isNoneEmpty(fileName)) {
                mongoTags.put("name", fileName);
            }
            if (reviewStatus != null && reviewStatus == ThirdPicReviewStatus.PERSON_REVIEW_PASS.getCode()) {
                mongoTags.put("reviewStatus", reviewStatus);
                mongoTags.put("reviewer", reviewer);
                mongoTags.put("reviewTime", new Date());
            }
            if (mongoTags.keySet().size() > 0) {
                collection.update(query, new BasicDBObject("$set", mongoTags), false, false,WriteConcern.JOURNALED);
            }
        } else {
            BasicDBObject upper = new BasicDBObject();

            BasicDBObject setDBList = new BasicDBObject();
            //单独处理性别
            Object gender = tags.get("gender");
            if (gender != null) {
                setDBList.put("tags.gender", gender);
            }

            if (ListUtil.isNotEmpty(tags.get("languages"))) {
                setDBList.put("tags.languages", tags.get("languages"));
            }
            if (ListUtil.isNotEmpty(tags.get("application"))) {
                setDBList.put("tags.application", tags.get("application"));
            }

            if (StringUtils.isNoneBlank(fileName)) {
                setDBList.put("name", fileName);
            }

            if (reviewStatus != null && reviewStatus == ThirdPicReviewStatus.PERSON_REVIEW_PASS.getCode()) {
                setDBList.put("reviewStatus", reviewStatus);
                setDBList.put("reviewer", reviewer);
                setDBList.put("reviewTime", new Date());
            }

            if (setDBList.size() > 0) {
                upper.append("$set", setDBList);
            }

            if (upper.size() > 0) {
                collection.update(query, upper, false, false);
            }
        }
    }

    @Override
    public void updateAdImageComment(String mongodbId, AdResourceComment comment) {

        DBObject query = new BasicDBObject("mongodbId", mongodbId);
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);

        BasicDBObject basicDBList = new BasicDBObject();
        if (comment != null && StringUtil.isNotEmpty(comment.getCommentBody())) {
            basicDBList.put("comments", JSON.toJSONString(comment));
        }
        if (basicDBList.keySet().size() > 0) {
            collection.update(query, new BasicDBObject("$push", basicDBList), false, false,WriteConcern.JOURNALED);
        }
    }

    @Override
    public void updateReviewStatus(String mongodbId, Integer reviewStatus, String adReviewFeedback, Integer machineCnt) {
        DBObject query = new BasicDBObject("mongodbId", mongodbId);
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject mongoTags = new BasicDBObject();
        mongoTags.put("reviewStatus", reviewStatus);
        if (StringUtil.isNotEmpty(adReviewFeedback)) {
            mongoTags.put("adReviewFeedback", adReviewFeedback);
        }
        if (machineCnt != null && machineCnt != 0) {
            mongoTags.put("machineCnt", machineCnt);
        }
        collection.update(query, new BasicDBObject("$set", mongoTags), false, false,WriteConcern.JOURNALED);
    }

    @Override
    public void batchUpdateReviewStatus(List<String> mongodbIds, Integer reviewStatus) {
        DBObject query = new BasicDBObject("mongodbId", new BasicDBObject("$in", mongodbIds));
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject mongoTags = new BasicDBObject();
        mongoTags.put("reviewStatus", reviewStatus);
        if (mongoTags.keySet().size() > 0) {
            collection.update(query, new BasicDBObject("$set", mongoTags), false, true,WriteConcern.JOURNALED);
        }

    }

    @Override
    public void removeByAccountIdAndDay(Long accountId, String day) {

        BasicDBObject query = new BasicDBObject();
        query.put("day", day);
        query.put("accountId", accountId);
        try {
            DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
            collection.remove(query);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public List<AdResourceStats> AdimagesAllListMonByimageBathId(String imageBathId) {

        if (StringUtils.isBlank(imageBathId)) {
            throw new IllegalArgumentException("参数错误，imageBathId不能为空。");
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("day", "1970-01-01");
        query.put("imageBathId", imageBathId);

        DBCursor cursor = collection.find(query).addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<AdResourceStats> adResourceStatses = new ArrayList<>();
        try {
            while (cursor.hasNext()) {
                DBObject next = cursor.next();
                AdResourceStats adResourceStats = JSONObject.parseObject(JSON.toJSONString(next), AdResourceStats.class);
                adResourceStatses.add(adResourceStats);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return adResourceStatses;
    }

    @Override
    public void batchUpdateReviewsTotalOfPageViews(List<AdResourceStats> adResourceStatses) {
        if (ListUtil.isEmpty(adResourceStatses)) {
            throw new IllegalArgumentException("参数错误，adResource。");
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        for (AdResourceStats adResourceStats : adResourceStatses) {
            DBObject query = new BasicDBObject("mongodbId", adResourceStats.getMongodbId());
            BasicDBObject document = new BasicDBObject();
            document.put("totoalPageViews", adResourceStats.getTotoalPageViews());
            if (document.keySet().size() > 0) {
                collection.update(query, new BasicDBObject("$set", document), false, false,WriteConcern.JOURNALED);
            }
        }
    }

    @Override
    public void updateCountPageViews(String userId, String mongodId, Long numOfPageViews, Long peopleNumOfPageViews) {
        if (StringUtils.isEmpty(mongodId)) {
            throw new IllegalArgumentException("参数错误，adResource。");
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject("mongodbId", mongodId);
        BasicDBObject document = new BasicDBObject();
        document.put("numOfPageViews", numOfPageViews);
        document.put("peopleNumOfPageViews", peopleNumOfPageViews);
        if (document.keySet().size() > 0) {
            collection.update(query, new BasicDBObject("$set", document), false, false,WriteConcern.JOURNALED);
        }
    }


    @Override
    public List<AdResourceStats> findUnReviewedImagesUploadByOutUi(int limit, String[] returnFields) {
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("day", "1970-01-01");
        query.put("from", 1);
        query.put("reviewStatus", 0);

        DBObject projectObject = new BasicDBObject();
        for (String returnField : returnFields) {
            projectObject.put(returnField, 1);
        }

        DBCursor cursor = collection.find(query, projectObject).addOption(Bytes.QUERYOPTION_NOTIMEOUT).limit(limit);
        if (cursor == null) {
            logger.warn("没有查询到第三方未审核的图片");
            return null;
        }
        List<AdResourceStats> result = new ArrayList<>();
        while (cursor.hasNext()) {
            DBObject next = cursor.next();
            if (next != null) {
                AdResourceStats adResourceStats = JSON.parseObject(JSON.toJSONString(next), AdResourceStats.class);
                result.add(adResourceStats);
            } else {
                logger.error("查询有空结果。");
            }
        }
        return result;
    }

    @Override
    public long countUnReviewedImagesUploadByOutUi() {
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("day", "1970-01-01");
        query.put("from", 1);
        query.put("reviewStatus", 0);
        return collection.count(query);
    }

    @Override
    public String getImageHashbyMd5(String md5) {
        if (StringUtils.isEmpty(md5)) {
            throw new IllegalArgumentException("参数错误，md5不能为空。");
        }
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("day", "1970-01-01");
        query.put("md5", md5);
        DBObject dbObject = collection.findOne(query);
        if (dbObject == null) {
            return null;
        }
        if (dbObject.get("imageHash") == null) {
            return "hashnull";
        }
        return dbObject.get("imageHash").toString();
    }

    @Override
    public List<AdResourceStats> getAdResourceImageHashIsNull() {
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject query = new BasicDBObject();
        query.put("day", "1970-01-01");
        query.put("imageHash", new BasicDBObject("$regex", ".*_imagemd5"));
        List<AdResourceStats> result = new ArrayList<>();
        DBCursor cursor = collection.find(query).addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        if (cursor == null) {
            logger.warn("没有查询到无hash的图片");
            return null;
        }
        while (cursor.hasNext()) {
            DBObject next = cursor.next();
            if (next != null) {
                AdResourceStats adResourceStats = JSON.parseObject(JSON.toJSONString(next), AdResourceStats.class);
                result.add(adResourceStats);
            } else {
                logger.error("查询有空结果。");
            }
        }
        return result;
    }

    @Override
    public void updateHashbyMongodbId(String mongodbId, String hash) {
        DBObject query = new BasicDBObject("mongodbId", mongodbId);
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject hashObj = new BasicDBObject();
        hashObj.put("imageHash", hash);
        collection.update(query, new BasicDBObject("$set", hashObj), false, false);
    }

    @Override
    public void deleteAdImages(String userId, List<String> mongodbIds) {
        DBObject query = new BasicDBObject("mongodbId", new BasicDBObject("$in", mongodbIds));
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject hashObj = new BasicDBObject();
        hashObj.put("deleteUser", userId);
        hashObj.put("deleteTime", new Date());
        hashObj.put("isDelete", 1);
        collection.update(query, new BasicDBObject("$set", hashObj), false, true);
    }

    @Override
    public void updateMap(String mongodbId, Map<String, Object> map) {
        DBObject query = new BasicDBObject("mongodbId", mongodbId);
        DBCollection collection = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY);
        DBObject hashObj = new BasicDBObject();
        for (Map.Entry<String, Object> stringObjectEntry : map.entrySet()) {
            String key = stringObjectEntry.getKey();
            Object value = stringObjectEntry.getValue();
            hashObj.put(key, value);
        }
        collection.update(query, new BasicDBObject("$set", hashObj), false, true,WriteConcern.JOURNALED);
    }

    @Override
    public List<AdResourceStats> getImageStatsGroupByBreakdowns(String imageHash, String since, String until, String breakdowns, boolean onlyNew, Set<Long> accountSet, Set<Long> fbConnectionObjectIds) {
        String breakdownsField = "";
        if (breakdowns.equals("age") || breakdowns.equals("gender")) {
            breakdownsField = "age,gender";
        } else {
            breakdownsField = breakdowns;
        }
        String[] breakdownsArr = new String[]{breakdowns};
        if (breakdowns.contains(",")) {
            breakdownsArr = breakdowns.split(",");
        }

        // 第一次Match，过滤统计数据
        BasicDBObject matchChildObject1 = new BasicDBObject()
                .append("breakdowns", breakdownsField)  // 细分字段过滤
                .append("imageHash", imageHash);  // imageHash过滤
        // 统计周期过滤
        if (onlyNew) {
            long dayDiff = 1;
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            try {
                dayDiff = DateUtil.dayDiff(simpleDateFormat.parse(since), simpleDateFormat.parse(until));
            } catch (Exception e) {
                logger.error("时间转换失败，startDay={},endDay={}", since, until);
            }
            matchChildObject1.put("day", "1970-01-02");
            matchChildObject1.put("dayRange", dayDiff);
        } else {
            matchChildObject1.append("day", new BasicDBObject().append("$gte", since).append("$lte", until));
        }
        if (!CollectionUtils.isEmpty(accountSet)) {
            matchChildObject1.append("accountId", new BasicDBObject("$in", accountSet)); // 统计账号过滤
        }
        if (!CollectionUtils.isEmpty(fbConnectionObjectIds)) {
            matchChildObject1.append("fbConnectionObjectId", new BasicDBObject("$in", fbConnectionObjectIds)); // 统计应用过滤
        }

        // 第一次group，根据指定条件分组汇总数据
        BasicDBObject idObjectForGroup1 = new BasicDBObject();
        BasicDBObject groupChildObject1 = new BasicDBObject()
                .append("_id", idObjectForGroup1);

        // 排序，根据细分字段排序
        DBObject sortChildObject = new BasicDBObject();

        for (String breakdown : breakdownsArr) {
            if (StringUtils.isNotEmpty(breakdown)) {
                idObjectForGroup1.put(breakdown, "$" + breakdown);
                sortChildObject.put("_id." + breakdown, 1);
            }
        }
        for (String field : SUM_FIELDS) {
            groupChildObject1.put(field, new BasicDBObject("$sum", "$" + field));
        }

        // 组合步骤，总共3步
        List<DBObject> aggregateObjects = new ArrayList<>();
        // 第一步，match，筛选数据
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        // 第二步，group，分组汇总数据
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));
        // 第三步，排序
        aggregateObjects.add(new BasicDBObject("$sort", sortChildObject));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<AdResourceStats> adResourceStatsList = new ArrayList<>();
        try {
            DBObject dbObject = null;
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                BasicDBObject idResult = (BasicDBObject) dbObject.get("_id");
                for (String breakdown : breakdownsArr) {
                    if (StringUtils.isNotEmpty(breakdown)) {
                        dbObject.put(breakdown, idResult.get(breakdown));
                    }
                }
                adResourceStatsList.add(JSONObject.parseObject(JSON.toJSONString(dbObject), AdResourceStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return adResourceStatsList;
    }

    @Override
    public List<UserAdResourceStats> getImageStatsGroupByUser() {
        // 第一次Match，过滤统计数据
        BasicDBObject groupChildObject1 = new BasicDBObject()
                .append("_id", new BasicDBObject("imageHash", "$imageHash"))
                .append("userIdSet", new BasicDBObject("$addToSet", "$userIdSet"));
        // 统计周期过滤
        for (String field : SUM_FIELDS) {
            groupChildObject1.put(field, new BasicDBObject("$sum", "$" + field));
        }

        BasicDBObject groupChildObject2 = new BasicDBObject()
                .append("_id",
                        new BasicDBObject("userId",
                                "$userIdSet"))
                .append("spent", new BasicDBObject("$sum", "$spent"))
                .append("effective_img_num",
                        new BasicDBObject("$sum",
                                new BasicDBObject("$cond",
                                        new BasicDBObject("if",
                                                new BasicDBObject("$gt", new Object[]{"$spent", 10000}))
                                                .append("then", 1).append("else", 0))))
                .append("hot_img_num",
                        new BasicDBObject("$sum",
                                new BasicDBObject("$cond",
                                        new BasicDBObject("if",
                                                new BasicDBObject("$gt", new Object[]{"$spent", 1000000}))
                                                .append("then", 1).append("else", 0))))
                .append("effective_img_spent",
                        new BasicDBObject("$sum",
                                new BasicDBObject("$cond",
                                        new BasicDBObject("if",
                                                new BasicDBObject("$gt", new Object[]{"$spent", 10000}))
                                                .append("then", "$spent").append("else", 0))));

        // 组合步骤，总共3步
        List<DBObject> aggregateObjects = new ArrayList<>();
        // 第一步，match，筛选数据
        aggregateObjects.add(new BasicDBObject("$match", new BasicDBObject("$or", new Object[]{
                new BasicDBObject("day", "1970-01-01"),
                new BasicDBObject("breakdowns", "all").append("day", new BasicDBObject("$gt", "1970-01-02"))})));
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));

        aggregateObjects.add(new BasicDBObject("$unwind", "$userIdSet"));
        // 第二步，group，分组汇总数据
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject2));
        // 第三步，排序
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = adResourceMongoDBClient.getCollection(MongoConstants.TABLE_NAME_AD_RESOURCE_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<UserAdResourceStats> adResourceStatsList = new ArrayList<>();
        try {
            DBObject dbObject = null;
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                BasicDBObject idResult = (BasicDBObject) dbObject.get("_id");
                dbObject.put("userId", ((BasicDBList) idResult.get("userId")).get(0));
                adResourceStatsList.add(JSONObject.parseObject(JSON.toJSONString(dbObject), UserAdResourceStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return adResourceStatsList;
    }
}
